package com.zyuc.spark.merge

import java.util
import java.util.Comparator

import com.zyuc.java.hdfs.{FileBox, HdfsFile}
import org.apache.hadoop.fs.{FileSystem, Path}

/**
  * Created by zhoucw on 18-8-24 下午5:15.
  */
class FileBoxService(filterFileSize: Long, boxVolume: Long) {

  /**
    * 按照文件大小降序排列, 返回HdfsFile对象的数组
    *
    * @param input
    * @param fileSystem
    * @return
    */
  private def getFilesSortBySize(input: String, fileSystem: FileSystem): Array[HdfsFile] = {

    val arrayList = new util.ArrayList[HdfsFile]()

    // 取出小文件
    val files = fileSystem.globStatus(new Path(input + "/*")).filter(_.getLen < filterFileSize).
      map(x => new HdfsFile(x.getPath.getName, x.getLen))


    // 自定义排序算法： 实现排序接口, 按照文件大小降序排列
    val comparator = new Comparator[HdfsFile] {

      override def compare(f1: HdfsFile, f2: HdfsFile): Int = {

        val deltaSize = f1.getFileSize - f2.getFileSize
        if (deltaSize > 0l) -1 else if (deltaSize < 0l) 1 else 0

      }
    }

    // 按照文件大小降序排列
    util.Arrays.sort(files, comparator)

    files
  }


  /**
    * 对文件进行分组装箱
    *
    * @param files
    * @return 返回箱子列表
    */

  private def groupFiles2Box(files: Array[HdfsFile]): util.ArrayList[FileBox] = {

    /////////////
    // 初始化
    ////////////
    var fileIndex = 0 // 文件序号
    var boxIndex = 0 // 箱子序号
    var fileBox = new FileBox
    val boxList = new util.ArrayList[FileBox]
    val newBoxList = new util.ArrayList[FileBox]

    try {
      // 遍历所有文件
      while (fileIndex < files.length) {

        val boxIterator = boxList.iterator

        ////////////////////////////
        // 遍历箱子, 按照箱子打开的时间遍历, 如果找到有剩余空间的箱子, 设置foundBox为true,
        // 如果没有找到合适的箱子, 需要打开一个新箱子
        ////////////////////////////
        var foundBox = false
        while (boxIterator.hasNext && !foundBox) {
          fileBox = boxIterator.next
          if (fileBox.getRemainder - files(fileIndex).getFileSize >= 0) {
            foundBox = true
          }
        }
        // 没有找到合适的箱子, 打开一个新箱子
        if (!foundBox) {
          fileBox = new FileBox

          // 获取箱子的容量
          fileBox.setRemainder(boxVolume)

          fileBox.setId(boxIndex)
          fileBox.setLinkedList(new util.LinkedList[HdfsFile])
          boxList.add(fileBox)
          boxIndex = boxIndex + 1
        }

        // 装箱子, 并修改箱子的属性
        val linkedList = fileBox.getLinkedList
        linkedList.add(files(fileIndex))
        fileBox.setLinkedList(linkedList)
        val reMainder = fileBox.getRemainder - files(fileIndex).getFileSize
        fileBox.setRemainder(reMainder)

        fileIndex = fileIndex + 1
      }

      // 过滤只有一个文件的箱子
      val boxListIterator = boxList.iterator()
      var index = 0
      while (boxListIterator.hasNext) {
        val box = boxListIterator.next()
        if (box.getLinkedList.size() > 1) {
          box.setId(index)
          newBoxList.add(box)
          index = index + 1
        }
      }
    } catch {
      case e: Exception => {
        e.printStackTrace()
      }
    }

    newBoxList
  }


  /**
    * 获取箱子列表
    *
    * @param input
    * @param fileSystem
    * @return
    */
  def getBoxList(input: String, fileSystem: FileSystem): util.ArrayList[FileBox] = {

    var boxList = new util.ArrayList[FileBox]()

    try{

      // 按照文件大小排序
      val files = getFilesSortBySize(input, fileSystem)

      // 将文件分组装箱, 返回箱子列表
      boxList = groupFiles2Box(files)

    }catch {

      case e:Exception=>{

        e.printStackTrace()
      }

    }

    boxList
  }

}
